package consys.social.memcached.dao.spymemcached;

import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import net.spy.memcached.AddrUtil;
import net.spy.memcached.BinaryConnectionFactory;
import net.spy.memcached.KetamaConnectionFactory;
import net.spy.memcached.MemcachedClient;
import net.spy.memcached.transcoders.Transcoder;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

/**
 * Memcached wrapper
 * Heavily inspired by Ashish Saxena
 * 
 * @author Petr Kunc
 */
public class SpyMemcachedImpl implements Memcached {

    private MemcachedClient client;
    private String[] servers;
    private int cacheOperationTimeoutSeconds = 1;
    private int defaultExpirationTimeSeconds = 3600;
    private Logger logger = Logger.getLogger(SpyMemcachedImpl.class);

    public SpyMemcachedImpl() {
        //do nothing - startUp later
    }

    public SpyMemcachedImpl(String[] servers) {
        setServers(servers);
    }

    public SpyMemcachedImpl(String type, String[] servers) {
        setServers(servers);
        startUp(type);
    }

    /**
     * Get a memcached client connected to server
     */
    public void startUp(String type) {
        try {
            if (type.equalsIgnoreCase("default")) {
                client = new MemcachedClient(AddrUtil.getAddresses(StringUtils.join(getServers(), ' ')));
            } else if (type.equalsIgnoreCase("binary")) {
                client = new MemcachedClient(new BinaryConnectionFactory(), AddrUtil.getAddresses(StringUtils.join(getServers(), ' ')));
            } else {
                client = new MemcachedClient(new KetamaConnectionFactory(), AddrUtil.getAddresses(StringUtils.join(getServers(), ' ')));
            }
        } catch (IOException e) {
            logger.warn("Could not start memcached");
        }
    }

    @Override
    public Object get(String key) {
        return get(key, null);
    }

    @Override
    public Object get(String key, Transcoder tc) {

        Object myObj = null;
        Future<Object> future = null;
        if (tc == null) {
            future = client.asyncGet(key);
        } else {
            future = client.asyncGet(key, tc);
        }

        try {
            myObj = future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            logger.error("Get not successful - timeout");
        } catch (Exception e) {
            logger.error("Get not successful" + e.getMessage());
        }
        return myObj;
    }

    @Override
    public boolean add(String key, Object value, int expiration) {
        return add(key, value, expiration, null);
    }

    @Override
    public boolean add(String key, Object value, int expiration, Transcoder tc) {
        Future<Boolean> future = null;
        if (tc == null) {
            future = client.add(key, expiration, value);
        } else {
            future = client.add(key, expiration, value, tc);
        }

        try {
            return future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            logger.error("Add not successful - timeout");
        } catch (Exception e) {
            logger.error("Add not successful - " + e.getMessage());
        }
        return false;
    }

    @Override
    public boolean delete(String key) {
        Future<Boolean> future = client.delete(key);
        try {
            return future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            logger.error("Delete not successful - timeout");
        } catch (Exception e) {
            logger.error("Delete not successful - " + e.getMessage());
        }
        return false;
    }

    @Override
    public boolean set(String key, Object value, int cacheTimeSeconds) {
        Future<Boolean> future = client.set(key, cacheTimeSeconds, value);
        try {
            return future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            logger.error("Set not successful - timeout");
        } catch (Exception e) {
            logger.error("Set not successful - " + e.getMessage());
        }
        return false;
    }

    @Override
    public boolean set(String key, Object o) {
        return set(key, o, getDefaultExpirationTimeSeconds());
    }

    @Override
    public boolean replace(String key, Object value, int expiration) {
        return replace(key, value, expiration, null);
    }

    @Override
    public boolean replace(String key, Object value, int expiration, Transcoder tc) {
        Future<Boolean> future = null;
        if (tc == null) {
            future = client.replace(key, expiration, value);
        } else {
            future = client.replace(key, expiration, value, tc);
        }
        try {
            return future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            logger.error("Replace not successful - timeout");
        } catch (Exception e) {
            logger.error("Replace not successful - " + e.getMessage());
        }
        return false;
    }

    /**
     * Array of distributed MemCached Server e.g "server1.mydomain.com:11211",
     * "server2.mydomain.com:11211", "server3.mydomain.com:11211"
     */
    public String[] getServers() {
        return servers;
    }

    public void setServers(String[] servers) {
        this.servers = servers;
    }

    public int getCacheOperationTimeoutSeconds() {
        return cacheOperationTimeoutSeconds;
    }

    public void setCacheOperationTimeoutSeconds(int cacheOperationTimeoutSeconds) {
        this.cacheOperationTimeoutSeconds = cacheOperationTimeoutSeconds;
    }

    public int getDefaultExpirationTimeSeconds() {
        return defaultExpirationTimeSeconds;
    }

    public void setDefaultExpirationTimeSeconds(int defaultExpirationTimeSeconds) {
        this.defaultExpirationTimeSeconds = defaultExpirationTimeSeconds;
    }

    @Override
    public void clear() {
        client.flush();
    }

    @Override
    public void shutdown() {
        client.shutdown();
    }

    public <T> Map<String, T> getBulk(Collection<String> keys, Transcoder<T> tc) {
        return client.getBulk(keys, tc);
    }

    public Map<String, Object> getBulk(Collection<String> keys) {
        return client.getBulk(keys);
    }

//  working but not needed
//    @Override
//    public long incrementByOne(String key) {
//        return increment(key, 1);
//    }
//
//    @Override
//    public long increment(String key, int by) {
//        long value = 0;
//        Future<Long> future = client.asyncIncr(key, by);
//        try {
//            value = future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
//        } catch (TimeoutException e) {
//            logger.error("Incr not successful - timeout");
//        } catch (Exception e) {
//            logger.error("Incr not successful" + e.getMessage());
//        }
//        return value;
//    }
//
//    @Override
//    public long decrementByOne(String key) {
//        return decrement(key, 1);
//    }
//
//    @Override
//    public long decrement(String key, int by) {
//        long value = 0;
//        Future<Long> future = client.asyncDecr(key, by);
//        try {
//            value = future.get(getCacheOperationTimeoutSeconds(), TimeUnit.SECONDS);
//        } catch (TimeoutException e) {
//            logger.error("Decr not successful - timeout");
//        } catch (Exception e) {
//            logger.error("Decr not successful" + e.getMessage());
//        }
//        return value;
//    }
}
